package com.headcaselabs.orchestra;

import java.util.Hashtable;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.headcaselabs.work.Monitor;
import com.headcaselabs.work.UnitOfWork;
/**
 * Provides utility function for launching monitored tasks and storing and retrieving Futures of tasks and their monitors.
 * @author Florian Leibert
 *
 */
public class JobUtils {
	public static final int MAX_CONCURRENT = 4;
	private static Hashtable<Object,Future<?>> mappings = new Hashtable<Object,Future<?>>();
	private static ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(MAX_CONCURRENT);
	private static Hashtable<String,UnitOfWork> taskid_job = new Hashtable<String,UnitOfWork>();
	private static List<Object> unitsOfWork = new LinkedList<Object>();
	private static List<String> cancelled_jobs = new LinkedList<String>();
	public static Object myLock = new Object();
	private static ReentrantReadWriteLock lock = new ReentrantReadWriteLock();

  private final static Log log_ =
    LogFactory.getLog(JobUtils.class.getName());
	
	/*public synchronized static Job getJob(String id) {
		return taskid_job.get(id);
	}
	
	public synchronized static void addJob(String id, Job job) {
		taskid_job.put(id, job);
		
	}
	
	public synchronized static Job removeJob(String id) {
		log_.debug("Removing job from mapping table with id:"+id+" \t"+taskid_job.get(id).toString());
		return taskid_job.remove(id);
	
	}*/
	
 	public synchronized static boolean runMonitoredJob(UnitOfWork uof) {
 		log_.info("Starting monitored job:"+uof.getId());
	 		Monitor m = new Monitor(uof, uof.getId()+"-monitor",uof.getMax_time());
	 		if (executor.getActiveCount()<MAX_CONCURRENT-2){
				uof.registerSupervisee(m);
				log_.debug("Submitting job for execution...");
				Future<Integer> fut = executor.submit(uof);
				uof.setFuture(fut);
				log_.debug("Submitted job for execution. Submitting monitor...");
				Future<Boolean> fut_mon = executor.submit(m);
				m.setFuture(fut_mon);
				log_.debug("Submitted monitor for execution.");
				//mappings.put(m,fut_mon);
				//mappings.put(uof,fut);
				log_.info("Started monitored job:"+uof.getId()+ " with monitor:"+m.getId());
				return true;
	 		} else {
	 			log_.debug("Not enough resources to start the job just yet.");
	 		}
	 		return false;
 		
	}
 	
 	public static boolean isReady() {
 		try {
 			lock.readLock().lock();
 			return (executor.getActiveCount()<MAX_CONCURRENT-2);
 		} finally {
 			lock.readLock().unlock();
 		}
 	}
 	
 	public static List<Object> getUnitsOfWork() {
 		try {
 			lock.readLock().lock();
 			return unitsOfWork;
 		} finally {
 			lock.readLock().unlock();
 		}
 	}
 	
 	
 	public static ScheduledThreadPoolExecutor getExecutor() {
 		try {
 			lock.readLock().lock();
 			return executor;
 		} finally {
 			lock.readLock().unlock();
 		}
 	}
 	
 	/*public static void addCancelledJob(String j) {
 	 synchronized(JobUtils.class) {
 			cancelled_jobs.add(j);
 	 }
 	}
 	
 	
 	public static synchronized boolean removeCancelledJob(String j) {
 		try {
 			lock.writeLock().lock();
 			return cancelled_jobs.remove(j);
 		} finally {
			lock.writeLock().unlock();
		}
 	}
 	
 	public static void removeAllCancelledJobs() {
 		try {
 			lock.writeLock().lock();
 			cancelled_jobs.clear();
 		} finally {
 			lock.writeLock().unlock();
 		}
 	}
	
 	public static Collection<String> getCancelledJobs() {
 		try {
 			lock.writeLock().lock();
 			cancelled_jobs.clear();
 			List<String> copy = new ArrayList<String>();
 			Collections.copy(copy,cancelled_jobs);
 			return copy;
 		} finally {
 			lock.writeLock().unlock();
 		}
 	}
 	*/
	
	

}
